\newpage

\subsection{Architecture}
ESJ operator aims at efficiently processing the streaming join in the large-scale distributed environment. To this end, we propose an architecture composed of the join engine and the peripherals as shown in Figure~\ref{fig:arch}.

\subsubsection{Join Engine}
To maximize the computational scalability, ESJ applies the dataflow-oriented processing model based on message passing. A series of join workers are connected in a cascading way by stream channels. For running multiple join tasks upon the same stream pair concurrently, a separated set of chained worker instances are assigned to the individual task. All worker instances belonging to the same worker share the stream pair so that the processing is conducted without duplicating the inputs.

In order to achieve globally workload balancing, ESJ enforces the criterion that a worker transfer part of the workload to its successor (with respect to the stream direction) if and only if its workload exceeds its successor's. This criterion guarantees that workloads are distributed to all workers evenly and automatically.

\subsubsection{Input Adapter and Load Shedder}
In order to make ESJ applicable to diverse inputs, the input adapter is adopted to convert the external data source into streaming input. Additionally, the load shedder is used to tolerate the fluctuation of input streams.

\subsubsection{Materialization}
The up-to-date outputs of the join engine are collected and kept in the memory buffer, while the obsolete ones are structured as snapshots and materialized to the persistent storage, i.e., database. As the the unit of materialization, a \textit{snapshot} is a set of records that share a common identifier. The committing of a snapshot could be either periodical or according to the punctuations~\cite{tkde03:Tucker} within the input streams. The committing timestamp is used as the snapshot identifier. 

\subsubsection{Query Proxy and Query Processor}
Clients retrieve the join results through queries. ESJ supports both the \textit{continuous query}, which requests up-to-date results infinitely, and the \textit{one-time query}, which requests the results within a time span. By applying the client-server model for query requesting and responding, multiple queries can be served simultaneously. The query proxy converts the clients' query requests into a stream events, and, reversely, the query response stream into the clients' format. The query processor answers the queries by interacting with the buffer manager and the database. Note that the query proxy is pluggable and the query processor is stateless, meaning that the query processor can handle a scaling number of query proxies.
